package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job01_UvTopnPage {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表,映射 kafka中的用户行为dwd明细宽表
        tenv.executeSql(
                " CREATE TABLE dwd_user_action       (                                    "+
                        " 	session_id STRING,                                            "+
                        " 	event_id STRING,                                              "+
                        " 	action_time BIGINT,                                           "+
                        " 	properties MAP<STRING,STRING>,                                "+
                        "   user_id bigint,                                               "+
                        "   create_time timestamp(3),                                     "+
                        "   page_type string,                                             "+
                        " 	page_business string,                                         "+
                        " 	url as properties['url'] ,                                    "+
                        " 	rt as to_timestamp_ltz(action_time,3) ,                       "+
                        " 	watermark for rt as rt                                        "+
                        " ) WITH (                                                        "+
                        "     'connector' = 'kafka',                                      "+
                        "     'topic' = 'dwd-user-action',                                "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',            "+
                        "     'properties.group.id' = 'g008',                             "+
                        "     'scan.startup.mode' = 'latest-offset',                      "+
                        "     'value.format' = 'json',                                    "+
                        "     'value.fields-include' = 'EXCEPT_KEY'                       "+
                        " )				                                                  "
        );


        // 建表，jdbc连接器映射mysql中的目标表

        tenv.executeSql(
                " CREATE TABLE sink_mysql  (                      "+
                        "   page_type    string,                          "+
                        "   rn           bigint,                          "+
                        "   window_start timestamp(3),                    "+
                        "   window_end   timestamp(3),                    "+
                        "   url          string,                          "+
                        "   page_uv      bigint,                          "+
                        "   PRIMARY KEY (page_type,rn) NOT ENFORCED       "+
                        " ) WITH (                                        "+
                        "    'connector' = 'jdbc',                        "+
                        "    'username' = 'root',                         "+
                        "    'password' = 'root',                         "+
                        "    'url' = 'jdbc:mysql://doitedu:3306/doit46',  "+
                        "    'table-name' = 'dashboard_uvtopn_page_5m'    "+
                        " )                                               "
        );


        // sql
        tenv.executeSql(
                        "INSERT INTO sink_mysql                                                                                    "+
                        "WITH tmp AS (                                                                                             "+
                        "     SELECT                                                                                                "+
                        "       window_start,                                                                                       "+
                        "     	window_end,                                                                                         "+
                        " 		page_type,                                                                                          "+
                        "     	url,                                                                                                "+
                        "     	count(distinct user_id) as page_uv                                                                  "+
                        "     FROM TABLE(                                                                                           "+
                        "         TUMBLE(TABLE dwd_user_action,DESCRIPTOR(rt),INTERVAL '5' MINUTE)                                  "+
                        "     )                                                                                                     "+
                        "     GROUP BY                                                                                              "+
                        "         window_start,                                                                                     "+
                        "     	window_end,                                                                                         "+
                        " 		page_type,                                                                                          "+
                        "     	url                                                                                                 "+
                        " )	                                                                                                        "+
                        "                                                                                                           "+
                        " SELECT                                                                                                    "+
                        " 	page_type,                                                                                              "+
                        " 	rn,                                                                                                     "+
                        "   window_start,                                                                                           "+
                        " 	window_end,                                                                                             "+
                        " 	url,                                                                                                    "+
                        " 	page_uv                                                                                                 "+
                        "                                                                                                           "+
                        " FROM (                                                                                                    "+
                        "     SELECT                                                                                                "+
                        "       window_start,                                                                                       "+
                        "     	window_end,                                                                                         "+
                        " 		page_type,                                                                                          "+
                        "     	url,                                                                                                "+
                        "     	page_uv,                                                                                            "+
                        "     	ROW_NUMBER() OVER(PARTITION BY window_start,window_end,page_type ORDER BY page_uv DESC) AS rn       "+
                        "     FROM tmp ) o                                                                                          "+
                        " WHERE rn<=2                                                                                              "
        );

    }

}
